package mq

import (
	"github.com/Shopify/sarama"
	"github.com/sirupsen/logrus"
)

func KafkaSend() {
	for {
		select {
		case msg := <-sendChan:
			pid, offset, err := producer.SendMessage(msg)
			if err != nil {
				logrus.Errorf("kafka: Send Topic %s  value %s is failed ", msg.Topic, msg.Value)
				continue
			}
			logrus.Infof("send message to kafka success, pid %v, offset: %v", pid, offset)
		}
	}
}

func KafkaRecd(topic string, handle func(*sarama.ConsumerMessage)) {
	// 拿到指定topic下面的所有分区列表
	partitionList, err := consumer.Partitions(topic) // 根据topic取到所有的分区
	if err != nil {
		logrus.Error("fail to get list of partition:", err)
		return
	}
	for partition := range partitionList { // 遍历所有分区并执行处理函数
		var pc sarama.PartitionConsumer
		pc, err = consumer.ConsumePartition(topic, int32(partition), sarama.OffsetNewest)

		go func(sarama.PartitionConsumer) {
			for {
				select {
				case msg := <-pc.Messages():
					handle(msg)
				}
			}
		}(pc)

	}
}
